package com.echo.im.mq.event.receiver;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.echo.im.mq.event.BaseEvent;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.core.ResolvableType;
import org.springframework.messaging.handler.annotation.Payload;

import java.nio.charset.StandardCharsets;
import java.util.Objects;

/**
 * 抽象消息接收器，用于处理从MQ接收的消息
 * 该类采用泛型方式处理不同类型的消息
 *
 * @param <T> 消息的具体类型
 * @author echo
 * @date 2024-08-15
 */
@Slf4j
public abstract class AbstractMessageReceiver<T> implements IMReceiver {

    private Class<T> type;

    private Class<T> getType(){
        if(Objects.isNull(type)){
            ResolvableType resolvableType = ResolvableType.forClass(getClass());
            this.type = (Class<T>) resolvableType.as(AbstractMessageReceiver.class).getGeneric(0).resolve();
            if (this.type == null) {
                throw new IllegalStateException("无法解析泛型类型");
            }
        }
        return type;
   }

    /**
     * 消息处理方法，当从MQ队列中接收到消息时调用
     * 负责解析消息内容，并调用具体的处理方法
     *
     * @param message 接收到的消息
     * @param channel MQ通道，用于确认或拒绝消息
     * @throws Exception 处理过程中可能出现的异常
     */
    @Override
    public void onMessage(@Payload Message message, Channel channel) throws Exception {
        // 将消息体转换为字符串
        String msg = new String(message.getBody(), StandardCharsets.UTF_8);
        // 记录接收到的消息
        log.info("xxx MQ 队列 xxx QUEUE 读取到消息 ::: [{}]", msg);
        // 获取消息的唯一标识
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {

            BaseEvent.EventMessage<T> eventMessage
                    = JSON.parseObject(msg, new TypeReference<>(getType()) {
            });

            T data = eventMessage.getData();

            // 调用具体的处理方法
            doReceive(data);

            // 确认消息处理成功
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            // 记录异常日志
            log.error("处理消息出现异常：{}", e.getMessage());
            log.error("RMQ_CHAT_TRAN_ERROR", e);
            log.error("NACK_MSG:{}", msg);
            // 拒绝消息，不重新放入队列
            channel.basicNack(deliveryTag, false, false);
        }
    }


    /**
     * 具体的消息处理方法，由子类实现
     *
     * @param data 具体类型的消息数据
     */
    public abstract void doReceive(T data);

}
